media_pp\elements\source/
file_demuxer.rs

1use std::{path::Path, sync::Arc, time::Duration};
2
3use crate::pp_log::{PpLog, pp_error, pp_info};
4use ffmpeg_next as ffmpeg;
5use thiserror::Error as ThisError;
6
7use crate::{
8    buffer::MediaBuffer,
9    bus::{Bus, BusEvent},
10    control::{ControlReceiver, drain_control},
11    element::{Element, ElementType, Source, SourceElement, element_pp_log},
12    pad::SrcPad,
13};
14
15/// Errors specific to `FileDemuxer`. Converts into the crate-wide `Error`
16/// via `?` (see [`crate::error::Error`]).
17#[derive(Debug, ThisError)]
18pub enum FileDemuxError {
19    #[error("ffmpeg error: {0}")]
20    Ffmpeg(#[from] ffmpeg::Error),
21}
22
23/// Metadata about one stream in an opened container, reported up front so
24/// callers can decide what to build downstream before the pipeline runs.
25#[derive(Debug, Clone, Copy)]
26pub struct StreamInfo {
27    pub index: usize,
28    pub kind: ffmpeg::media::Type,
29}
30
31/// Demuxes a file, exposing one src pad per container stream (indexed the
32/// same way as `StreamInfo::index`). Linking a pad "selects" that stream;
33/// leaving it unlinked just drops its packets. Real demuxer I/O is
34/// blocking, so this is meant to be run as the pipeline's source thread.
35///
36/// Fan-out (e.g. routing video and audio to separate branches) needs no
37/// separate "Tee" element here — it's just a matter of linking more than
38/// one of these pads.
39pub struct FileDemuxer {
40    pp_log: PpLog,
41    name: Arc<str>,
42    input: ffmpeg::format::context::Input,
43    pads: Vec<SrcPad>,
44    /// One packet read ahead of `run`'s own loop, set only by `seek` —
45    /// peeking a packet right after `Input::seek` is how it learns where
46    /// playback actually landed (see `seek`'s docs), and that packet still
47    /// needs to be delivered, not discarded, so it's stashed here for
48    /// `run`'s next iteration to pick up instead of reading a fresh one.
49    pending: Option<(usize, ffmpeg::Packet)>,
50}
51
52impl FileDemuxer {
53    /// Opens the file and returns it alongside every stream it contains,
54    /// so the caller can inspect them (count, media type, ...) before
55    /// deciding which of `src_pads()` to link.
56    pub fn open(
57        name: impl Into<String>,
58        path: impl AsRef<Path>,
59    ) -> Result<(Self, Vec<StreamInfo>), FileDemuxError> {
60        let input = ffmpeg::format::input(&path)?;
61
62        let streams: Vec<StreamInfo> = input
63            .streams()
64            .map(|s| StreamInfo {
65                index: s.index(),
66                kind: s.parameters().medium(),
67            })
68            .collect();
69
70        let pads = streams
71            .iter()
72            .map(|s| SrcPad::new(format!("src_{}", s.index)))
73            .collect();
74
75        let name: Arc<str> = name.into().into();
76        let pp_log = element_pp_log(ElementType::FileDemuxer, &name, None);
77        pp_info!(
78            pp_log: &pp_log,
79            "opened: path={}, {} stream(s)",
80            path.as_ref().display(),
81            streams.len()
82        );
83        Ok((
84            Self {
85                name,
86                pp_log,
87                input,
88                pads,
89                pending: None,
90            },
91            streams,
92        ))
93    }
94
95    /// Codec parameters for one of this file's streams — what you need to
96    /// construct a matching [`crate::elements::SwDecoder`] for it.
97    pub fn stream_parameters(&self, index: usize) -> Option<ffmpeg::codec::Parameters> {
98        self.stream(index).map(|s| s.parameters())
99    }
100
101    /// The unit decoded frame timestamps for this stream are expressed in —
102    /// what you need to construct a matching [`crate::elements::Pacer`] for
103    /// it.
104    pub fn stream_time_base(&self, index: usize) -> Option<ffmpeg::Rational> {
105        self.stream(index).map(|s| s.time_base())
106    }
107
108    fn stream(&self, index: usize) -> Option<ffmpeg::format::stream::Stream<'_>> {
109        self.input.streams().find(|s| s.index() == index)
110    }
111}
112
113impl Element for FileDemuxer {
114    fn name(&self) -> Arc<str> {
115        self.name.clone()
116    }
117
118    fn element_type(&self) -> ElementType {
119        ElementType::FileDemuxer
120    }
121
122    fn pp_log(&self) -> &PpLog {
123        &self.pp_log
124    }
125
126    fn pp_log_mut(&mut self) -> &mut PpLog {
127        &mut self.pp_log
128    }
129}
130
131impl Source for FileDemuxer {
132    fn src_pads(&mut self) -> &mut [SrcPad] {
133        &mut self.pads
134    }
135}
136
137impl SourceElement for FileDemuxer {
138    fn run(&mut self, control: &ControlReceiver, bus: &Bus) -> crate::error::Result<()> {
139        pp_info!(self, "started");
140        // Deliberately re-creates `self.input.packets()` fresh every
141        // iteration (cheap — it's just a short-lived wrapper, not a
142        // stateful cursor of its own) instead of holding one `for` loop's
143        // iterator across the whole function, the way this used to read.
144        // That iterator borrows `input` for as long as it's alive; `Seek`
145        // needs `drain_control` to be able to call `self.seek()` — a
146        // *second* mutable borrow of `input` — in between reads, which a
147        // single loop-spanning iterator would rule out.
148        loop {
149            if drain_control(control, self, bus)?.stopped {
150                // Stop: abandon in place, no final Eos.
151                pp_info!(self, "stopped");
152                return Ok(());
153            }
154            // `seek` (called from within `drain_control`, above) already
155            // consumed one packet to find out where it landed — deliver
156            // that before reading a fresh one, or it'd be silently lost.
157            let next = match self.pending.take() {
158                Some(next) => Some(next),
159                None => self.input.packets().next().map(|(s, p)| (s.index(), p)),
160            };
161            let Some((index, packet)) = next else {
162                break;
163            };
164            if let Some(pad) = self.pads.get_mut(index) {
165                // A downstream failure drops just this one packet — same
166                // "report, don't die" contract `Queue`'s worker gives a
167                // failing `Sink` — rather than ending this whole source
168                // thread over it. `Pipeline::stop` is how a caller who
169                // decides an error is fatal actually ends things.
170                if let Err(error) = pad.push(MediaBuffer::Packet(Arc::new(packet))) {
171                    bus.post(
172                        &self.pp_log,
173                        BusEvent::Error {
174                            element_type: ElementType::FileDemuxer,
175                            name: self.name.clone(),
176                            error,
177                        },
178                    );
179                }
180            }
181        }
182        for pad in self.pads.iter_mut() {
183            pad.push_eos(&self.pp_log)?;
184        }
185        pp_info!(self, "event=eos phase=source_completed outcome=ok");
186        Ok(())
187    }
188
189    fn seek(&mut self, target: Duration) -> crate::error::Result<Duration> {
190        // `Input::seek` takes microseconds (`AV_TIME_BASE` units) when
191        // seeking the whole container (stream index -1, which is what it
192        // uses internally) rather than one specific stream — an unbounded
193        // range (`..`) just means "as close to `ts` as ffmpeg can manage",
194        // no extra min/max constraint. In practice that means *backward*
195        // to the nearest keyframe at or before `target`: never forward,
196        // and never onto a non-keyframe, since either would leave nothing
197        // downstream can decode/remux from. A sparse-keyframe file can
198        // make that keyframe well before `target` — e.g. a single
199        // 10-second file with keyframes only at 0s and 8.3s means every
200        // `target` under 8.3s lands back at 0s.
201        let ts = target.as_micros().min(i64::MAX as u128) as i64;
202        self.input.seek(ts, ..).inspect_err(|error| {
203            pp_error!(self, "seek to {target:?} failed: {error}");
204        })?;
205
206        // `avformat_seek_file` only reports success/failure, not where it
207        // landed — the one way to find out is to read the next packet and
208        // look at its own timestamp. That packet is real data (not a
209        // probe to throw away), so it's stashed in `pending` for `run`'s
210        // next iteration instead of being dropped here.
211        match self.input.packets().next() {
212            Some((stream, packet)) => {
213                let time_base = stream.time_base();
214                let landed = packet
215                    .pts()
216                    .or_else(|| packet.dts())
217                    .map(|ts| ts_to_duration(ts, time_base))
218                    .unwrap_or(Duration::ZERO);
219                self.pending = Some((stream.index(), packet));
220                Ok(landed)
221            }
222            // Nothing left to read right after seeking (`target` at/past
223            // EOF) — there's no packet to learn a real position from, so
224            // just report the request back as-is.
225            None => Ok(target),
226        }
227    }
228}
229
230fn ts_to_duration(ts: i64, time_base: ffmpeg::Rational) -> Duration {
231    let secs = ts as f64 * f64::from(time_base.numerator()) / f64::from(time_base.denominator());
232    Duration::from_secs_f64(secs.max(0.0))
233}
234
235#[cfg(test)]
236mod tests {
237    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
238
239    use super::*;
240    use crate::control;
241    use crate::test_support::try_test_video;
242
243    struct CountingSink {
244        pp_log: PpLog,
245        count: Arc<AtomicUsize>,
246        saw_eos: Arc<AtomicBool>,
247    }
248
249    impl Element for CountingSink {
250        fn name(&self) -> Arc<str> {
251            "counting-sink".into()
252        }
253
254        fn element_type(&self) -> ElementType {
255            ElementType::Other
256        }
257
258        fn pp_log(&self) -> &PpLog {
259            &self.pp_log
260        }
261
262        fn pp_log_mut(&mut self) -> &mut PpLog {
263            &mut self.pp_log
264        }
265    }
266
267    impl crate::element::Sink for CountingSink {
268        fn consume(&mut self, buf: MediaBuffer) -> crate::error::Result<()> {
269            match buf {
270                MediaBuffer::Eos => self.saw_eos.store(true, Ordering::SeqCst),
271                _ => {
272                    self.count.fetch_add(1, Ordering::SeqCst);
273                }
274            }
275            Ok(())
276        }
277
278        fn control(&mut self, _msg: crate::control::ControlMsg) -> crate::error::Result<()> {
279            Ok(())
280        }
281    }
282
283    #[test]
284    fn open_reports_stream_parameters_for_a_valid_index_and_none_out_of_range() {
285        let Some(path) = try_test_video() else { return };
286        let (demuxer, streams) = FileDemuxer::open("demux", &path).expect("open test video");
287        let video = streams
288            .iter()
289            .find(|s| s.kind == ffmpeg::media::Type::Video)
290            .expect("test video has a video stream");
291
292        assert!(demuxer.stream_parameters(video.index).is_some());
293        assert!(demuxer.stream_time_base(video.index).is_some());
294
295        let out_of_range = streams.len() + 1;
296        assert!(
297            demuxer.stream_parameters(out_of_range).is_none(),
298            "an out-of-range stream index must report nothing, not panic"
299        );
300        assert!(demuxer.stream_time_base(out_of_range).is_none());
301    }
302
303    /// Drives `FileDemuxer::run` directly (no `Pipeline`) to prove the
304    /// basic contract on its own: every packet on a linked pad's stream
305    /// arrives, and running off the end of the file delivers a final
306    /// `Eos` rather than just stopping silently.
307    #[test]
308    fn run_delivers_every_packet_on_a_linked_pad_then_eos() {
309        let Some(path) = try_test_video() else { return };
310        let (mut demuxer, streams) = FileDemuxer::open("demux", &path).expect("open test video");
311        let video = streams
312            .iter()
313            .find(|s| s.kind == ffmpeg::media::Type::Video)
314            .expect("test video has a video stream");
315
316        let count = Arc::new(AtomicUsize::new(0));
317        let saw_eos = Arc::new(AtomicBool::new(false));
318        demuxer.src_pads()[video.index].link(Box::new(CountingSink {
319            count: count.clone(),
320            saw_eos: saw_eos.clone(),
321            pp_log: element_pp_log(ElementType::Other, "counting-sink", None),
322        }));
323
324        let (bus, bus_rx) = Bus::new();
325        let (_tx, rx) = control::channel();
326        demuxer
327            .run(&rx, &bus)
328            .expect("run must reach eos cleanly, not error");
329
330        assert!(
331            count.load(Ordering::SeqCst) > 0,
332            "expected at least one packet delivered to the linked pad"
333        );
334        assert!(
335            saw_eos.load(Ordering::SeqCst),
336            "expected an Eos once the file is exhausted"
337        );
338        drop(bus);
339        assert!(
340            bus_rx.iter().all(|e| !matches!(e, BusEvent::Error { .. })),
341            "run must not report any errors demuxing a well-formed file"
342        );
343    }
344}
media_pp\elements\source/file_demuxer.rs

media_pp\elements\source/
file_demuxer.rs
